﻿use actix_web::{body::to_bytes, http::StatusCode, HttpResponse};
use sqlx::{Executor, PgPool, PgTransaction};
use uuid::Uuid;

use super::IdempotencyKey;

async fn get_saved_response(
    pool: &PgPool,
    idempotency_key: &IdempotencyKey,
    user_id: Uuid,
) -> Result<Option<HttpResponse>, anyhow::Error> {
    let saved_response = sqlx::query!(
        r#"SELECT RESPONSE_STATUS_CODE as "response_status_code!",
		RESPONSE_HEADERS AS "response_headers!: Vec<HeaderPairRecord>",
		RESPONSE_BODY AS "response_body!"
		FROM idempotency
		WHERE USER_ID=$1 AND IDEMPOTENCY_KEY = $2"#,
        user_id,
        idempotency_key.as_ref()
    )
    .fetch_optional(pool)
    .await?;

    if let Some(r) = saved_response {
        let status_code = StatusCode::from_u16(r.response_status_code.try_into()?)?;
        let mut response = HttpResponse::build(status_code);
        for HeaderPairRecord { name, value } in r.response_headers {
            response.append_header((name, value));
        }
        Ok(Some(response.body(r.response_body)))
    } else {
        Ok(None)
    }
}

#[derive(Debug, sqlx::Type)]
#[sqlx(type_name = "header_pair")]
struct HeaderPairRecord {
    name: String,
    value: Vec<u8>,
}

pub async fn save_response(
    mut transaction: PgTransaction<'_>,
    idempotency_key: &IdempotencyKey,
    user_id: Uuid,
    http_response: HttpResponse,
) -> Result<HttpResponse, anyhow::Error> {
    let (response_head, body) = http_response.into_parts();
    // MessageBody::Error 不是 Send + Sync
    // 因此，它与 anyhow 不能很好兼容
    let body = to_bytes(body).await.map_err(|e| anyhow::anyhow!("{}", e))?;
    let status_code = response_head.status().as_u16() as i16;
    let headers = {
        let mut h = Vec::with_capacity(response_head.headers().len());
        for (name, value) in response_head.headers() {
            let name = name.as_str().to_owned();
            let value = value.as_bytes().to_owned();
            h.push(HeaderPairRecord { name, value });
        }
        h
    };

    // SQL 查询
    let query = sqlx::query_unchecked!(
        r#"UPDATE idempotency
		SET
			RESPONSE_STATUS_CODE = $3,
			RESPONSE_HEADERS = $4,
			RESPONSE_BODY = $5
		WHERE USER_ID = $1 AND IDEMPOTENCY_KEY = $2"#,
        user_id,
        idempotency_key.as_ref(),
        status_code,
        headers,
        body.as_ref()
    );
    transaction.execute(query).await?;
    transaction.commit().await?;

    // 使用 .map_into_boxed_body 方法将 HttpResponse<Bytes> 转换为 HttpResponse<BoxBody>
    let http_response = response_head.set_body(body).map_into_boxed_body();
    Ok(http_response)
}

#[allow(clippy::large_enum_variant)]
pub enum NextAction {
    StartProcessing(PgTransaction<'static>), // 稍后使用返回的事务
    ReturnSavedResponse(HttpResponse),
}

pub async fn try_processing(
    pool: &PgPool,
    idempotency_key: &IdempotencyKey,
    user_id: Uuid,
) -> anyhow::Result<NextAction> {
    let mut transaction = pool.begin().await?;
    let query = sqlx::query!(
        r#"INSERT INTO idempotency
		( USER_ID, IDEMPOTENCY_KEY, CREATED_AT)
		VALUES ($1, $2, now())
		ON CONFLICT DO NOTHING"#,
        user_id,
        idempotency_key.as_ref()
    );
    let n_inserted_rows = transaction.execute(query).await?.rows_affected();

    if n_inserted_rows > 0 {
        Ok(NextAction::StartProcessing(transaction))
    } else {
        let saved_response = get_saved_response(pool, idempotency_key, user_id)
            .await?
            .ok_or_else(|| anyhow::anyhow!("We expected a saved response, we didn't find it"))?;
        Ok(NextAction::ReturnSavedResponse(saved_response))
    }
}
